package org.afk.tools.io;

import java.io.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import org.afk.tools.log.*;

/**
 * Collects Files and puts them into its queue. Access Files with getResult().
 *
 * @author axel
 */
public class FileCollector extends Thread
{
	public static FileCollector createMultiFileCollector(final File root, final int fixedThreadPoolCount)
	{
		return new FileCollector(root, -1, Executors.newFixedThreadPool(fixedThreadPoolCount), new LinkedBlockingQueue<File>(), new ConsoleErrorHandler());
	}

	public static FileCollector createSimpleFileCollector(final File root)
	{
		return new FileCollector(root, -1, Executors.newSingleThreadScheduledExecutor(), new LinkedBlockingQueue<File>(), new ConsoleErrorHandler());
	}

	private final File root;
	private final int maxDepth;
	private boolean cancel;

	private final ExecutorService executors;

	private final BlockingQueue<File> result;
	private final ErrorHandler errorHandler;
	private Throwable t;

	private final AtomicLong active = new AtomicLong();

	public FileCollector(final File root, final int maxDepth, final ExecutorService executors, final LinkedBlockingQueue<File> result,
			final ErrorHandler errorHandler)
	{
		super();
		this.root = root;
		this.maxDepth = maxDepth;
		this.executors = executors;
		this.result = result;
		this.errorHandler = errorHandler;
	}

	private void addFilesToQueue(final File[] subFiles) throws InterruptedException
	{
		for (final File file : subFiles)
			while (!this.result.offer(file, 1, TimeUnit.SECONDS))
				if (this.cancel)
					break;
	}

	private void recurse(final File dir, final int currentLevel) throws InterruptedException
	{
		this.active.incrementAndGet();
		this.executors.submit(new Runnable()
		{

			private void processDirectories(final int nextLevel, final File[] subFiles) throws InterruptedException
			{
				for (final File file : subFiles)
					if (file.isDirectory())
						recurse(file, nextLevel);
			}

			@Override
			public void run()
			{

				try
				{
					final int nextLevel = currentLevel + 1;
					final File[] subFiles = dir.listFiles();
					addFilesToQueue(subFiles);
					if ((FileCollector.this.maxDepth < 0) || (nextLevel <= FileCollector.this.maxDepth))
						processDirectories(nextLevel, subFiles);
				}
				catch (final InterruptedException e)
				{
					FileCollector.this.errorHandler.warning(e);
					FileCollector.this.t = e;
				}
				finally
				{
					FileCollector.this.active.decrementAndGet();
				}
			}
		});

	}

	private void waitForEmptyExecutors() throws InterruptedException
	{
		while (this.active.intValue() > 0)
			if (!this.cancel)
				Thread.sleep(150);
	}

	public void cancel()
	{
		this.cancel = true;
	}

	public BlockingQueue<File> getResult()
	{
		return this.result;
	}

	public Throwable getT()
	{
		return this.t;
	}

	public boolean isTerminated()
	{
		return this.executors.isTerminated();
	}

	@Override
	public void run()
	{
		try
		{
			if (this.root.isDirectory())
				recurse(this.root, 0);
			else
				this.result.add(this.root);

			waitForEmptyExecutors();
			if (this.cancel)
				this.executors.shutdownNow();
			else
				this.executors.shutdown();
		}
		catch (final Throwable t)
		{
			setT(t);
		}
	}

	public void setT(final Throwable t)
	{
		this.t = t;
	}

}
